

package com.example.embed.calcite.service;

import lombok.Getter;
import org.apache.calcite.DataContext;
import org.apache.calcite.adapter.java.JavaTypeFactory;
import org.apache.calcite.linq4j.AbstractEnumerable;
import org.apache.calcite.linq4j.Enumerable;
import org.apache.calcite.linq4j.Enumerator;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.schema.ScannableTable;
import org.apache.calcite.schema.impl.AbstractTable;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.sql.type.SqlTypeUtil;
import org.apache.calcite.util.Pair;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * 内存表实现类，用于在Apache Calcite中提供基于内存的数据表支持
 * <p>
 * 该类实现了ScannableTable接口，允许Calcite引擎扫描和查询内存中的数据。
 * 主要功能包括：
 * - 管理表结构（列定义）
 * - 存储表数据
 * - 提供数据扫描能力
 * - 支持动态类型映射（Java类型到SQL类型）
 * </p>
 * 
 * @author zwg
 * @date 2021/10/13
 */
public class MemoryTable extends AbstractTable implements ScannableTable {

    /**
     * 表名
     */
    @Getter
    public final String tableName;

    /**
     * 表的列定义列表
     */
    @Getter
    private List<Column> columns;

    /**
     * 表的数据行，每行是一个键值对映射
     */
    private List<Map<String, Object>> items;

    /**
     * 创建只有表名的内存表实例
     * 
     * @param tableName 表名
     */
    public MemoryTable(String tableName) {
        this.tableName = tableName;
    }

    /**
     * 设置表数据并自动提取列信息
     * 
     * @param items 表数据行列表
     */
    public void setItems(List<Map<String, Object>> items) {
        this.items = items;
        this.columns = extractColumn(items.get(0));
    }

    /**
     * 创建完整的内存表实例
     * 
     * @param tableName 表名
     * @param columns 列定义列表
     * @param items 表数据行列表
     */
    public MemoryTable(String tableName,
                       List<Column> columns, List<Map<String, Object>> items) {
        this.tableName = tableName;
        this.columns = columns;
        this.items = items;
    }

    /**
     * 创建内存表实例，自动从数据提取列信息
     * 
     * @param tableName 表名
     * @param items 表数据行列表
     */
    public MemoryTable(String tableName, List<Map<String, Object>> items) {
        this(tableName, extractColumn(items.get(0)), items);
    }


    /**
     * 从数据行中提取列信息
     * 
     * @param item 单个数据行（键值对映射）
     * @return 提取出的列定义列表
     */
    public static List<Column> extractColumn(Map<String, Object> item) {
        List<Column> cols = new ArrayList<>();
        for (Map.Entry<String, Object> objectEntry : item.entrySet()) {
            Column column = new Column();
            column.setCode(objectEntry.getKey());
            column.setName(objectEntry.getKey());
            column.setValueType(FieldValueTypeConverter.convertJavaTypeByValue(objectEntry.getValue()));
            cols.add(column);
        }
        return cols;
    }
/*
    @Override
    public RelDataType getRowType(RelDataTypeFactory relDataTypeFactory) {
        JavaTypeFactory typeFactory = (JavaTypeFactory) relDataTypeFactory;
        //column name
        List<String> names = new ArrayList<>();
        //column type
        List<RelDataType> types = new ArrayList<>();
        for (Column col : columns) {
            names.add(col.getCode());
            RelDataType relDataType = typeFactory.createJavaType(col.getValueType().getClazz());
            relDataType = SqlTypeUtil.addCharsetAndCollation(relDataType, typeFactory);
            types.add(relDataType);
        }
        return typeFactory.createStructType(Pair.zip(names, types));
    }*/

    /**
     * 获取表的行类型定义，将Java类型映射到SQL类型
     * <p>
     * 实现了AbstractTable中的抽象方法，为Calcite引擎提供表的元数据信息
     * </p>
     * 
     * @param typeFactory 类型工厂，用于创建SQL类型
     * @return 表的行类型定义
     */
    @Override
    public RelDataType getRowType(RelDataTypeFactory typeFactory) {
        if (items.isEmpty()) {
            return typeFactory.builder().build();
        }

        RelDataTypeFactory.FieldInfoBuilder fieldInfo = typeFactory.builder();

        Map<String, Object> firstRow = items.get(0);
        for (Map.Entry<String, Object> entry : firstRow.entrySet()) {
            String fieldName = entry.getKey();
            Object value = entry.getValue();

            // 映射 Java 类型到 SQL 类型
            RelDataType sqlType;
            if (value instanceof Integer) {
                sqlType = typeFactory.createSqlType(SqlTypeName.INTEGER);
            } else if (value instanceof String) {
                sqlType = typeFactory.createSqlType(SqlTypeName.VARCHAR);
            } else if (value instanceof LocalDateTime) {
                sqlType = typeFactory.createSqlType(SqlTypeName.TIMESTAMP);
            } else {
                sqlType = typeFactory.createSqlType(SqlTypeName.ANY);
            }

            fieldInfo.add(fieldName, sqlType);
        }

        return fieldInfo.build();
    }

    /**
     * 扫描表数据，返回可枚举的结果集
     * <p>
     * 实现了ScannableTable接口，允许Calcite引擎扫描表中的所有数据
     * </p>
     * 
     * @param dataContext 数据上下文
     * @return 表数据的可枚举结果集
     */
    @Override
    public Enumerable<Object[]> scan(DataContext dataContext) {
        return new AbstractEnumerable<Object[]>() {
            @Override
            public Enumerator<Object[]> enumerator() {
                return new MemoryEnumerator(items, columns);
            }
        };
    }
}
